/*
 * ./ip_reassembly-static --vdev="event_sw0" -s 0x10
 *                     +-----------------+
 *                     | +-------------+ |
 *        +-------+    | |    flow 0   | |   
 *        |Packet |    | +-------------+ |  
 *        |event  |    | +-------------+ |
 *        |       |    | |    flow 1   | |port_link(port0, queue0)
 *        +-------+    | +-------------+ |     |     +--------+
 *        +-------+    | +-------------+ o-----v-----o        |dequeue +------+
 *        |Crypto |    | |    flow n   | |           | event  +------->|Core 0|
 *        |work   |    | +-------------+ o----+      | port 0 |        |      |
 *        |done ev|    |  event queue 0  |    |      +--------+        +------+
 *        +-------+    +-----------------+    |   事件端口本身不选择流
 *        +-------+                           |流控制是通过队列分配实现的
 *        |Timer  |    +-----------------+    |      +--------+
 *        |expiry |    | +-------------+ |    +------o        |dequeue +------+
 *        |event  |    | |    flow 0   | o-----------o event  +------->|Core 1|
 *        +-------+    | +-------------+ |      +----o port 1 |        |      |
 *       Event enqueue | +-------------+ |      |    +--------+        +------+
 *     o-------------> | |    flow 1   | |      |
 *        enqueue(     | +-------------+ |      |
 *        queue_id,    |                 |      |    +--------+        +------+
 *        flow_id,     | +-------------+ |      |    |        |dequeue |Core 2|
 *        sched_type,  | |    flow n   | o-----------o event  +------->|      |
 *        event_type,  | +-------------+ |      |    | port 2 |        +------+
 *        subev_type,  |  event queue 1  |      |    +--------+
 *        event)       +-----------------+      |    +--------+
 *                                              |    |        |dequeue +------+
 *        +-------+    +-----------------+      |    | event  +------->|Core n|
 *        |Core   |    | +-------------+ o-----------o port n |        |      |
 *        |(SW)   |    | |    flow 0   | |      |    +--------+        +--+---+
 *        |event  |    | +-------------+ |      |                         |
 *        +-------+    | +-------------+ |      |                         |
 *            ^        | |    flow 1   | |      |                         |
 *            |        | +-------------+ o------+                         |
 *            |        | +-------------+ |                                |
 *            |        | |    flow n   | |                                |
 *            |        | +-------------+ |                                |
 *            |        |  event queue n  |                                |
 *            |        +-----------------+                                |
 *            |                                                           |
 *            +-----------------------------------------------------------+
 *
 */


#include <rte_eal.h>
#include <rte_eventdev.h>
#include <rte_event_eth_rx_adapter.h>
#include <rte_mempool.h>
#include <rte_ethdev.h>
#include <rte_malloc.h>

#define MAX_EVENTS    128
#define MEMPOOL_SIZE  8192
#define SCHED_QUEUE_ID 0
#define SCHED_PORT_ID  0
struct rte_mempool* mbuf_pool = NULL;
int port = 0;
void init_port()
{
    int ret;

    uint16_t nb_rxd = 1024;
    uint16_t nb_txd = 1024;
    struct rte_eth_dev_info dev_info;
    struct rte_eth_rxconf rxq_conf;
    struct rte_eth_txconf txq_conf;
    struct rte_eth_conf local_port_conf  = {
        .txmode = {
            .mq_mode = ETH_MQ_TX_NONE,
        },
    };
    ret = rte_eth_dev_info_get(port, &dev_info);
    if (ret != 0)
        rte_exit(EXIT_FAILURE,
            "Error during getting device (port %u) info: %s\n",
            port, strerror(-ret));

    if (dev_info.tx_offload_capa & DEV_TX_OFFLOAD_MBUF_FAST_FREE)
        local_port_conf.txmode.offloads |=
            DEV_TX_OFFLOAD_MBUF_FAST_FREE;
    ret = rte_eth_dev_configure(port, 1, 1, &local_port_conf);
    if (ret < 0)
        rte_exit(EXIT_FAILURE, "Could not configure port%u (%d)\n",
                    (unsigned)port, ret);

    ret = rte_eth_dev_adjust_nb_rx_tx_desc(port, &nb_rxd, &nb_txd);
    if (ret < 0)
        rte_exit(EXIT_FAILURE, "Could not adjust number of descriptors "
                "for port%u (%d)\n", (unsigned)port, ret);

    rxq_conf = dev_info.default_rxconf;
    rxq_conf.offloads = local_port_conf.rxmode.offloads;
    ret = rte_eth_rx_queue_setup(port, 0, nb_rxd,
        rte_eth_dev_socket_id(port), &rxq_conf, mbuf_pool);
    if (ret < 0)
        rte_exit(EXIT_FAILURE, "Could not setup up RX queue for "
                "port%u (%d)\n", (unsigned)port, ret);

    txq_conf = dev_info.default_txconf;
    txq_conf.offloads = local_port_conf.txmode.offloads;
    ret = rte_eth_tx_queue_setup(port, 0, nb_txd,
        rte_eth_dev_socket_id(port), &txq_conf);
    if (ret < 0)
        rte_exit(EXIT_FAILURE, "Could not setup up TX queue for "
                "port%u (%d)\n", (unsigned)port, ret);

    ret = rte_eth_dev_start(port);
    if (ret < 0)
        rte_exit(EXIT_FAILURE, "Could not start port%u (%d)\n",
                        (unsigned)port, ret);

    if (1) {
        ret = rte_eth_promiscuous_enable(port); // 开启混杂接收模式
        if (ret != 0)
            rte_exit(EXIT_FAILURE,
                "Could not enable promiscuous mode for port%u: %s\n",
                port, rte_strerror(-ret));
    }
}
int main(int argc, char **argv)
{
    int ret;

    // 初始化DPDK环境
    ret = rte_eal_init(argc, argv);
    if (ret < 0)
        rte_exit(EXIT_FAILURE, "Failed to initialize EAL\n");

    int dev_id = 0;
    const struct rte_event_dev_config config = {
        .nb_event_queues = 3,
        .nb_event_ports = 6,
        .nb_events_limit  = 4096,
        .nb_event_queue_flows = 1024,
        .nb_event_port_dequeue_depth = 128,
        .nb_event_port_enqueue_depth = 128,
    };
    int err = rte_event_dev_configure(dev_id, &config);
    if (err)
        rte_exit(EXIT_FAILURE, "EventPort配置阶段失败! errCode: %d\n", err);



    struct rte_event_queue_conf atomic_conf = {
        .schedule_type = RTE_SCHED_TYPE_ATOMIC,
        .priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
        .nb_atomic_flows = 1024,
        .nb_atomic_order_sequences = 1024,
    };
    struct rte_event_queue_conf single_link_conf = {
        .event_queue_cfg = RTE_EVENT_QUEUE_CFG_SINGLE_LINK,
    };

    int atomic_q_1 = 0;
    int atomic_q_2 = 1;
    int single_link_q = 2;
    err = rte_event_queue_setup(dev_id, atomic_q_1, &atomic_conf);
    if (err)
        rte_exit(EXIT_FAILURE, "EventQueue0启动失败! errCode: %d\n", err);

    err = rte_event_queue_setup(dev_id, atomic_q_2, &atomic_conf);
    if (err)
        rte_exit(EXIT_FAILURE, "EventQueue1启动失败! errCode: %d\n", err);

    err = rte_event_queue_setup(dev_id, single_link_q, &single_link_conf);
    if (err)
        rte_exit(EXIT_FAILURE, "EventQueue2启动失败! errCode: %d\n", err);


    struct rte_event_port_conf rx_conf = {
        .dequeue_depth = 128,
        .enqueue_depth = 128,
        .new_event_threshold = 1024,
    };
    struct rte_event_port_conf worker_conf = {
        .dequeue_depth = 16,
        .enqueue_depth = 64,
        .new_event_threshold = 4096,
    };
    struct rte_event_port_conf tx_conf = {
        .dequeue_depth = 128,
        .enqueue_depth = 128,
        .new_event_threshold = 4096,
    };

    int rx_port_id = 0;
    int worker_port_id;
    err = rte_event_port_setup(dev_id, rx_port_id, &rx_conf);
    if (err)
        rte_exit(EXIT_FAILURE, "EventPort0启动失败! errCode: %d\n", err);


    for (worker_port_id = 1; worker_port_id <= 4; worker_port_id++) {
        err = rte_event_port_setup(dev_id, worker_port_id, &worker_conf);
        if (err)
            rte_exit(EXIT_FAILURE, "EventPort%d启动失败! errCode: %d\n", worker_port_id, err);

    }

    int tx_port_id = 5;
    err = rte_event_port_setup(dev_id, tx_port_id, &tx_conf);
    if (err)
        rte_exit(EXIT_FAILURE, "EventPort5启动失败! errCode: %d\n", err);

    tx_port_id = 5;
    uint8_t atomic_qs[] = {0, 1};
    uint8_t single_link_qs[] = {2};
    uint8_t priority = RTE_EVENT_DEV_PRIORITY_NORMAL;
    for (worker_port_id = 1; worker_port_id <= 4; worker_port_id++) {
        int links_made = rte_event_port_link(dev_id, worker_port_id, atomic_qs, NULL, 2);
    }
    int links_made = rte_event_port_link(dev_id, tx_port_id, single_link_qs, &priority, 1);


    err = rte_event_dev_start(dev_id);
    if (err)
        rte_exit(EXIT_FAILURE, "EventDev启动失败! errCode: %d\n", err);

    // 创建内存池（修正 data_room_size）
    mbuf_pool = rte_pktmbuf_pool_create(
        "mbuf_pool", 8192 * 16, 32, 0,
        RTE_MBUF_DEFAULT_BUF_SIZE,  // 使用默认值 2176
        (int)rte_socket_id()
    );
    if (mbuf_pool == NULL)
        rte_exit(EXIT_FAILURE, "Could not initialise mbuf pool\n");
    int nb_sys_ports = rte_eth_dev_count_avail();
    if (nb_sys_ports == 0)
        rte_exit(EXIT_FAILURE, "No supported Ethernet device found\n");
    init_port();

     //rte_event_enqueue_burst();
     //rte_event_dequeue_burst();
     struct rte_event ev;
     struct rte_event ev_out;
     struct rte_mbuf *pkts_burst[1];
     while (1) {
         // 发送事件
         ev.queue_id = atomic_q_1;
         ev.priority = RTE_EVENT_DEV_PRIORITY_NORMAL;
         int nb_rx = rte_eth_rx_burst(port, 0, pkts_burst, 1);
         if (nb_rx == 0) {
             // 增加重试逻辑或调整调度策略
             rte_pause(); // 短暂等待
             continue;
         }
         ev.mbuf = pkts_burst[0];
         uint16_t dequeued = rte_event_enqueue_burst(dev_id,1, &ev, 1);
         if (dequeued == 0) {
             // 增加重试逻辑或调整调度策略
             rte_pause(); // 短暂等待
             continue;
         }
         // 处理事件
         dequeued = rte_event_dequeue_burst(dev_id, 1,&ev_out, 1,0);
         if (dequeued != 0) {
             struct rte_mbuf* a = ev_out.mbuf;
             struct rte_ether_hdr *ethhdr = rte_pktmbuf_mtod(a, struct rte_ether_hdr *);
             if (ethhdr->ether_type != rte_cpu_to_be_16(RTE_ETHER_TYPE_IPV4)) {
                 // 跳过非IPv4数据包
                 continue;
             }
             struct rte_ipv4_hdr *iphdr = rte_pktmbuf_mtod_offset(a, struct rte_ipv4_hdr *, sizeof(struct rte_ether_hdr));
             if (iphdr->next_proto_id != IPPROTO_UDP) {
                 // 跳过非UDP数据包
                 continue;
             }
             struct rte_udp_hdr *udphdr = (struct rte_udp_hdr *)(iphdr + 1);
             char *udp_data = (char *)(udphdr + 1);

             uint16_t udp_data_len = ntohs(udphdr->dgram_len) - sizeof(struct rte_udp_hdr);
             // 将 UDP 数据转为小端并输出
             uint32_t src_ip = iphdr->src_addr;
             uint32_t src_ip_host = ntohl(src_ip);
             printf("收到数据来自: ");
             printf("Source IP: %u.%u.%u.%u  Data: ",
                    (src_ip_host >> 24) & 0xFF,
                    (src_ip_host >> 16) & 0xFF,
                    (src_ip_host >> 8)  & 0xFF,
                    (src_ip_host)       & 0xFF);
             for (int i = 0; i < udp_data_len; i += 1) {
                 char word = *(udp_data + i);
                 printf("%c", word);
             }
             printf("\n");
         }
     }

}
// --vdev="event_sw0" -s 0x10